package com.github.kezhenxu94.crawler.source.rss;

import com.github.kezhenxu94.data.common.Lifecycle;
import com.github.kezhenxu94.data.sink.Sink;
import com.github.kezhenxu94.data.source.Source;

/**
 * Created by kezhenxu on 1/8/17.
 *
 * @author kezhenxu (kezhenxu94@163.com)
 */
public class Channel<T> implements Lifecycle {

    protected Sink<T> sink;
    protected Source<T> source;


    @Override
    public void setup() throws Exception {
        this.sink.setup();
        this.source.setup();
    }

    @Override
    public void start() throws Exception {
        this.sink.start();
        this.source.start();
        while (isRunning()) {
            this.sink.put(this.source.take());
        }
    }

    @Override
    public void close() throws Exception {
        try {
            this.source.close();
        } finally {
            this.sink.close();
        }
    }

    @Override
    public boolean isRunning() {
        return this.sink.isRunning() || this.source.isRunning();
    }

    public Sink<T> getSink() {
        return sink;
    }

    public void setSink(Sink<T> sink) {
        this.sink = sink;
    }

    public Source<T> getSource() {
        return source;
    }

    public void setSource(Source<T> source) {
        this.source = source;
    }
}
